#pragma once
#include<QThread>
#include<QSemaphore>
#include"SwitchRecordStore.h"
#include"IO.h"
using namespace std;


class AsyncIncreWriteThread : public QThread{
private:

    //单例模式，这块不需要懒加载。异步增量写入线程一定需要。
    static AsyncIncreWriteThread* asyncIncreWriteThread;

    SwitchRecordStore* recordStore;

    QSemaphore semaphore;

    int32_t batchSize;

    atomic_int32_t unInCreWriteCount;  //保证并发操作安全  信号量会有强制写入操作，而这个数据可靠

    AsyncIncreWriteThread(const int32_t& size){
        this->recordStore = new SwitchRecordStore();
        this->batchSize = size;
        this->unInCreWriteCount = 0;
    }

    //析构函数里面都是同步释放 仅在最头部控制异步
    ~AsyncIncreWriteThread(){
        MemUtil::clearPtrMem(recordStore);
    }

    void addComitInfo(const vector<pair<ResourceInfo, vector<Record*>>>& upDateInfo, const vector<pair<ResourceInfo, vector<IndexRecord*>>>& indexUpDateInfo, const vector<pair<ResourceInfo, vector<uint64_t>>>& bitMapUpDateInfo){
        const int32_t& recCount = recordStore->addComitInfo(upDateInfo, indexUpDateInfo, bitMapUpDateInfo);
        //依据的是record个数
        unInCreWriteCount.fetch_add(recCount);  //先更新个数再释放信号量
        semaphore.release(recCount);
    }

    void forceIncreWriteToDisk(){
        semaphore.release(batchSize);   //释放所需的信号量 强制开始异步写入
        while(unInCreWriteCount.load() != 0);    //非阻塞等待完成 如果本来就为0会立刻返回
    }

    bool isResUnIncreWriteToDisk(const ResourceInfo& resInfo){
        return recordStore->isResUnIncreWriteToDisk(resInfo);
    }

protected:

    virtual void run(){
        while(true){
            //这里获取信号量不严格 外界调用相关函数释放所需信号量会强制刷进磁盘
            semaphore.acquire(batchSize);
            if(unInCreWriteCount.load() == 0){ continue; }  //无需异步写入 跳过直接等待下一次异步写
            //根据记录写磁盘 并更新同步进度
            IOUtil::updateSyncProgress(recordStore->increWriteToDisk() * Record::getRecordSize());
            unInCreWriteCount.store(0);
        }
    }

public:

    static void addComitInfoToWriteThread(const vector<pair<ResourceInfo, vector<Record*>>>& upDateInfo, const vector<pair<ResourceInfo, vector<IndexRecord*>>>& indexUpDateInfo, const vector<pair<ResourceInfo, vector<uint64_t>>>& bitMapUpDateInfo){
        asyncIncreWriteThread->addComitInfo(upDateInfo, indexUpDateInfo, bitMapUpDateInfo);
    }

    static void threadForceIncreWriteToDisk(){
        asyncIncreWriteThread->forceIncreWriteToDisk();
    }

    static bool isThreadUnIncreWriteToDisk(const ResourceInfo& resInfo){
        return asyncIncreWriteThread->isResUnIncreWriteToDisk(resInfo);
    }

    static AsyncIncreWriteThread* getAsyncIncreWriteThread(){
        return asyncIncreWriteThread;
    }

    static void init(const int32_t& size){
        asyncIncreWriteThread = new AsyncIncreWriteThread(size);
        asyncIncreWriteThread->start();
    }

};
